package com.bwie.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bwie.es.EsLog;
import com.bwie.pojo.TbLog;
import com.bwie.service.TbLogService;
import com.bwie.mapper.TbLogMapper;
import com.bwie.utils.R;
import com.bwie.vo.LogPageVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;

/**
* @author 28242
* @description 针对表【tb_log】的数据库操作Service实现
* @createDate 2025-05-05 13:40:04
*/
@Service
@Slf4j
public class TbLogServiceImpl extends ServiceImpl<TbLogMapper, TbLog>
    implements TbLogService{

    @Autowired
    TbLogMapper tbLogMapper;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    @PostConstruct
    public void readlogEs(){

        IndexOperations indexOps = elasticsearchRestTemplate.indexOps(EsLog.class);
        if (!indexOps.exists()){
            System.out.println("索引不存在创建索引");
            indexOps.create();

            Document mapping = indexOps.createMapping();
            indexOps.putMapping(mapping);
        }
    }


    @KafkaListener(topics = "topic-log")
    public void recv(String message, Acknowledgment acknowledgment){
        log.info("Kafka接收到日志消息:"+message);

        TbLog tbLog = JSON.parseObject(message, TbLog.class);

        tbLogMapper.insert(tbLog);



        EsLog esLog = BeanUtil.toBean(tbLog, EsLog.class);
        elasticsearchRestTemplate.save(esLog);


        acknowledgment.acknowledge();
    }

    @Override
    public R getLogList(LogPageVo logPageVo) {

        //构造分页参数

        Integer pageNum=0;
        if (logPageVo.getPageNum()!=null && logPageVo.getPageNum()>0){
            pageNum=logPageVo.getPageNum()-1;
        }
        PageRequest pageRequest = PageRequest.of(pageNum, logPageVo.getPageSize());



        //构造搜索条件
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
//        boolQueryBuilder.must(
//                QueryBuilders.matchAllQuery()
//        );
        if (StringUtils.isNotEmpty(logPageVo.getKeyWord())){
            boolQueryBuilder.should(
                    QueryBuilders.matchQuery("logContent",logPageVo.getKeyWord())
            );
        }


        //开始查询
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        nativeSearchQueryBuilder.withPageable(pageRequest).withQuery(boolQueryBuilder);
        SearchHits<EsLog> searchHits = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), EsLog.class);


        //读取查询结果

        long total = searchHits.getTotalHits();

        ArrayList<EsLog> records = new ArrayList<>();
        for (SearchHit<EsLog> searchHit : searchHits.getSearchHits()) {
            EsLog esLog = searchHit.getContent();
            records.add(esLog);
        }

        //返回查询结构

        HashMap<String, Object> map = new HashMap<>();

        map.put("total",total);
        map.put("records",records);

        return R.OK(map);
    }
}




